package cn.com.guage.flink.transformation;

import java.util.Arrays;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import cn.com.guage.flink.domain.UserAction;

/**
 * 
 * @author yangdechao filter算子用法
 *
 */
public class FilterTransformation2 {

	 public static void main(String[] args) throws Exception{

	        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	        // 输入: 用户行为。某个用户在某个时刻点击或浏览了某个商品，以及商品的价格。
	        DataStreamSource<UserAction> source = env.fromCollection(Arrays.asList(
	                new UserAction("userID1", 1293984000, "click", "productID1", 10),
	                new UserAction("userID2", 1293984001, "browse", "productID2", 8),
	                new UserAction("userID1", 1293984002, "click", "productID1", 10)
	        ));

	        // 过滤: 过滤出用户ID为userID1的用户行为
	        SingleOutputStreamOperator<UserAction> result = source.filter(new FilterFunction<UserAction>() {
	            /**
				 * 
				 */
				private static final long serialVersionUID = 113895690552383426L;

				public boolean filter(UserAction value) throws Exception {
	                return value.getUserId().equals("userID1");
	            }
	        });
	        result.print();

	        env.execute();

	    }
}
